package com.ruoyi.flink.util.write;

import com.alibaba.fastjson.JSONObject;
import com.ruoyi.flink.baseutil.DateUtil;
import com.ruoyi.flink.util.db.ClickHouseLink;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.List;
import java.util.UUID;

/**
 * @Description: 向 ClickHouse写入数据 (在实际项目中还是按一般的写法来吧, 这种写法可能会导致线程不安全)
 * @ClassName: ClickHouseWrite
 * @author syc
 * @date: 2022年5月26日 下午7:38:06
 */
public class ClickHouseWrite extends RichSinkFunction<JSONObject> {

	private static final long serialVersionUID = 1L;

	Connection connection = null;
	String sql;
    String url = null;
    String user = null;
    String password = null;
	/**
	 *	下标,字段名,类型,值是否需要生成
	 */
	List<Tuple4<Integer,String,String,Boolean>> indexFieldTypeIfvalue;

	public ClickHouseWrite(String sql, List<Tuple4<Integer,String,String,Boolean>> indexFieldTypeIfvalue) {
		this.sql = sql;
		this.indexFieldTypeIfvalue = indexFieldTypeIfvalue;  
	}

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
        if(url != null && user != null && password != null) {
        	connection = ClickHouseLink.getConn();
        } else {
        	connection = ClickHouseLink.getConn(url, user, password);
        }  
	}

	@Override
	public void close() throws Exception {
		super.close();
		if (connection != null) {
			connection.close();
		}
	}

	@Override
	public void invoke(JSONObject json, Context context) throws Exception {
		PreparedStatement preparedStatement = connection.prepareStatement(sql);
		for (int i = 0; i < indexFieldTypeIfvalue.size(); i++) {
			switch (indexFieldTypeIfvalue.get(i).f2) {
			case "int":
				if(indexFieldTypeIfvalue.get(i).f3) {
					// 默认先给个0吧
					preparedStatement.setInt(indexFieldTypeIfvalue.get(i).f0,0);
				} else {
					preparedStatement.setInt(indexFieldTypeIfvalue.get(i).f0,json.getIntValue(indexFieldTypeIfvalue.get(i).f1));
				}
				break;
			case "string":
				if(indexFieldTypeIfvalue.get(i).f3) {
					// 生成一格uuid吧
					preparedStatement.setString(indexFieldTypeIfvalue.get(i).f0,UUID.randomUUID().toString().replace("-", ""));
				} else {
					preparedStatement.setString(indexFieldTypeIfvalue.get(i).f0,json.getString(indexFieldTypeIfvalue.get(i).f1));
				}
				break;
			case "double":
				if(indexFieldTypeIfvalue.get(i).f3) {
					// 默认先给个0吧
				    preparedStatement.setDouble(indexFieldTypeIfvalue.get(i).f0,0D);
				} else {
				    preparedStatement.setDouble(indexFieldTypeIfvalue.get(i).f0,json.getDoubleValue(indexFieldTypeIfvalue.get(i).f1));
				}
				break;
			case "long":
				if(indexFieldTypeIfvalue.get(i).f3) {
				    preparedStatement.setLong(indexFieldTypeIfvalue.get(i).f0, 0L);
				} else {    
				    preparedStatement.setLong(indexFieldTypeIfvalue.get(i).f0, json.getLongValue(indexFieldTypeIfvalue.get(i).f1));
				}
				break;
			case "float":
				if(indexFieldTypeIfvalue.get(i).f3) {
				    preparedStatement.setFloat(indexFieldTypeIfvalue.get(i).f0, 0F);
				} else {    
				    preparedStatement.setFloat(indexFieldTypeIfvalue.get(i).f0, json.getFloatValue(indexFieldTypeIfvalue.get(i).f1));
				}
				break;
			case "bigDecimal":
				if(indexFieldTypeIfvalue.get(i).f3) {
				    preparedStatement.setBigDecimal(indexFieldTypeIfvalue.get(i).f0, new BigDecimal(0));
				} else {    
				    preparedStatement.setBigDecimal(indexFieldTypeIfvalue.get(i).f0, json.getBigDecimal(indexFieldTypeIfvalue.get(i).f1));
				}
				break;		
			case "date":
				if(indexFieldTypeIfvalue.get(i).f3) {
				    preparedStatement.setDate(indexFieldTypeIfvalue.get(i).f0, new java.sql.Date(System.currentTimeMillis()));
				} else {  
				    preparedStatement.setDate(indexFieldTypeIfvalue.get(i).f0, new java.sql.Date(json.getDate(indexFieldTypeIfvalue.get(i).f1).getTime()));
				} 
				break;
			case "time":
				if(indexFieldTypeIfvalue.get(i).f3) {
				    preparedStatement.setString(indexFieldTypeIfvalue.get(i).f0, DateUtil.getDateTimeChar());
				} else {  
				    preparedStatement.setTime(indexFieldTypeIfvalue.get(i).f0, new java.sql.Time(json.getDate(indexFieldTypeIfvalue.get(i).f1).getTime()));
				}
				break;
			default:
				// 这里应该是要报错的或者放个Object?
				preparedStatement.setString(indexFieldTypeIfvalue.get(i).f0,json.getString(indexFieldTypeIfvalue.get(i).f1));
				break;
			}			
		}

		preparedStatement.execute();
		// connection.commit();
		System.out.println("完毕");
	}
	
	
	
	
	
	
	

}
